package com.bw.day0709;


import com.bw.lei.gd3_1;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Flink SQL实时数据分析程序
 * 功能：从Kafka消费用户行为数据，进行5秒窗口聚合计算，输出各类用户行为指标
 *
 * @Author:
 * @Date: 2025/07/04/11:52
 */
public class gd3_1_dws {
    public static void main(String[] args) throws Exception {
        // 1. 创建Flink流处理执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度为1（生产环境应根据实际情况调整）
        env.setParallelism(1);

        // 2. 创建Table API环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 3. 定义Kafka数据源表（使用Flink SQL DDL）
        tableEnv.executeSql("CREATE TABLE KafkaTable (\n" +
                "    shop_id STRING,          -- 店铺ID\n" +
                "    uid STRING,              -- 用户ID\n" +
                "    behavior STRING,         -- 用户行为类型(pay/view等)\n" +
                "    gz_id STRING,            -- 关注ID\n" +
                "    zbj_id STRING,           -- 主播ID\n" +
                "    dsp_id STRING,           -- DSP广告ID\n" +
                "    tw_id STRING,            -- 推文ID\n" +
                "    `3D_id` STRING,          -- 3D内容ID\n" +
                "    tsl_id STRING,           -- 特殊标签ID\n" +
                "    is_new BIGINT,           -- 是否新用户(0:老用户,1:新用户)\n" +
                "    page_id string,          -- 页面ID\n" +
                "    during_time double,      -- 页面停留时间(秒)\n" +
                "    item STRING,             -- 商品/内容项ID\n" +
                "    item_type STRING,        -- 商品类型(sku_id等)\n" +
                "    ts BIGINT,               -- 事件时间戳(毫秒)\n" +
                "    time_ltz AS TO_TIMESTAMP_LTZ(ts, 3),  -- 将时间戳转为TIMESTAMP_LTZ类型\n" +
                "    WATERMARK FOR time_ltz AS time_ltz - INTERVAL '10' SECOND" +  // 定义水印，允许10秒延迟
                ") WITH (\n" +
                "  'connector' = 'kafka',     -- 使用Kafka连接器\n" +
                "  'topic' = 'gd3_dwd_l',     -- Kafka主题名称\n" +
                "  'properties.bootstrap.servers' = 'hadoop102:9092',  -- Kafka集群地址\n" +
                "  'properties.group.id' = 'testGroup',  -- 消费者组ID\n" +
                "  'scan.startup.mode' = 'latest-offset',  -- 从最新偏移量开始消费\n" +
                "  'format' = 'json'          -- 数据格式为JSON\n" +
                ")");

        // 4. 创建5秒滚动窗口视图
        tableEnv.executeSql(
                "CREATE VIEW windowed_data AS\n" +
                        "SELECT \n" +
                        "    window_start,           -- 窗口开始时间\n" +
                        "    window_end,             -- 窗口结束时间\n" +
                        "    shop_id,                -- 店铺ID\n" +
                        "    item,                   -- 商品/内容项\n" +
                        "    item_type,              -- 商品类型\n" +
                        "    behavior,               -- 用户行为\n" +
                        "    uid,                    -- 用户ID\n" +
                        "    tsl_id,                 -- 特殊标签ID\n" +
                        "    is_new,                 -- 是否新用户\n" +
                        "    gz_id,                  -- 关注ID\n" +
                        "    zbj_id,                 -- 主播ID\n" +
                        "    tw_id,                  -- 推文ID\n" +
                        "    dsp_id,                 -- DSP广告ID\n" +
                        "    `3D_id`,                -- 3D内容ID\n" +
                        "    during_time,            -- 停留时间\n" +
                        "    page_id                 -- 页面ID\n" +
                        "FROM TABLE(\n" +
                        "    TUMBLE(TABLE KafkaTable, DESCRIPTOR(time_ltz), INTERVAL '5' seconds)\n" +  // 5秒滚动窗口
                        ")");

        // 5. 执行聚合查询，计算各类用户行为指标
        Table table = tableEnv.sqlQuery("SELECT\n" +
                " CAST(window_start AS STRING) AS windowStart,  -- 窗口开始时间(字符串格式)\n" +
                " CAST(window_end AS STRING) AS windowEnd,      -- 窗口结束时间(字符串格式)\n" +
                " shop_id as shopId,                            -- 店铺ID\n" +
                " COUNT(DISTINCT CASE WHEN shop_id IS NOT NULL THEN uid END) AS shopUv,  -- 店铺UV(独立访客)\n" +
                " CASE WHEN item_type='sku_id' THEN item END AS item,  -- 商品ID(仅当类型为sku_id)\n" +
                " COUNT(DISTINCT CASE WHEN item_type='sku_id' THEN uid END) AS itemUv,  -- 商品UV\n" +
                " COUNT(DISTINCT CASE WHEN behavior='pay' THEN uid END) AS payUv,       -- 支付用户数\n" +
                " COUNT(DISTINCT uid) AS uv,                     -- 总UV\n" +
                " COUNT(*) AS pv,                                -- 总PV(页面访问量)\n" +
                " tsl_id as tslId,                               -- 特殊标签ID\n" +
                " CASE WHEN COUNT(DISTINCT uid) > 0 THEN COUNT(*)/COUNT(DISTINCT uid) ELSE 0 END AS avgPv,  -- 人均PV\n" +
                " avg(during_time) AS avgDuringTime,             -- 平均停留时间\n" +
                " MAX(CASE WHEN is_new=0 THEN is_new END) AS oldd,  -- 老用户标识\n" +
                " COUNT(DISTINCT CASE WHEN is_new=0 THEN uid END) AS oldUv,  -- 老用户数\n" +
                " MAX(CASE WHEN is_new=1 THEN is_new END) AS neww,  -- 新用户标识\n" +
                " COUNT(DISTINCT CASE WHEN is_new=1 THEN uid END) AS newUv,  -- 新用户数\n" +
                " gz_id as gzId,                                -- 关注ID\n" +
                " COUNT(DISTINCT CASE WHEN gz_id IS NOT NULL THEN uid END) AS gzUv,  -- 关注用户数\n" +
                " zbj_id as zbjId,                              -- 主播ID\n" +
                " COUNT(DISTINCT CASE WHEN zbj_id IS NOT NULL THEN uid END) AS zbjUv,  -- 主播间用户数\n" +
                " tw_id as twId,                                -- 推文ID\n" +
                " COUNT(DISTINCT CASE WHEN tw_id IS NOT NULL THEN uid END) AS twUv,    -- 推文用户数\n" +
                " dsp_id as dspId,                              -- DSP广告ID\n" +
                " COUNT(DISTINCT CASE WHEN dsp_id IS NOT NULL THEN uid END) AS dspUv,  -- 广告用户数\n" +
                " `3D_id` as dddId,                             -- 3D内容ID\n" +
                " COUNT(DISTINCT CASE WHEN `3D_id` IS NOT NULL THEN uid END) AS dddIdUv,  -- 3D内容用户数\n" +
                " CASE WHEN page_id='good_detail' THEN page_id END AS pageId,  -- 商品详情页ID\n" +
                " cast(COUNT(DISTINCT CASE WHEN page_id='good_detail' THEN uid END) as String) AS pageUv  -- 商品详情页UV\n" +
                "FROM windowed_data  \n" +
                "GROUP BY \n" +  // 分组字段
                "    window_start, \n" +
                "    window_end,\n" +
                "    shop_id,\n" +
                "    item,\n" +
                "    item_type,\n" +
                "    tsl_id,\n" +
                "    is_new,\n" +
                "    gz_id,\n" +
                "    zbj_id,\n" +
                "    tw_id,\n" +
                "    dsp_id,\n" +
                "    `3D_id`,\n" +
                "    page_id");

        // 6. 将结果表转换为DataStream
        DataStream<gd3_1> appendStream = tableEnv.toAppendStream(table, gd3_1.class);

        // 7. 打印结果到控制台(生产环境应替换为实际Sink)
        appendStream.print("===============>");

        // 8. 执行Flink作业
        env.execute();
    }
}
